package simpledb;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * Knows how to compute some aggregate over a set of StringFields.
 */
public class StringAggregator implements Aggregator {

    private static final long serialVersionUID = 1L;

    /**
     * Aggregate constructor
     * @param gbfield the 0-based index of the group-by field in the tuple, or NO_GROUPING if there is no grouping
     * @param gbfieldtype the type of the group by field (e.g., Type.INT_TYPE), or null if there is no grouping
     * @param afield the 0-based index of the aggregate field in the tuple
     * @param what aggregation operator to use -- only supports COUNT
     * @throws IllegalArgumentException if what != COUNT
     */
     private int gbIndex;
     private Type gbfieldtype;
     private int aIndex;
     private Op what;
     private TupleDesc tupleDesc;
     private Map<Field,Integer> gbfield2agval;

    public StringAggregator(int gbfield, Type gbfieldtype, int afield, Op what) {
        // some code goes here

    }

    public StringAggregator(int gbIndex, Type gbfieldtype, int aIndex, Op what, TupleDesc tupleDesc) {
        this.gbIndex = gbIndex;
        this.gbfieldtype = gbfieldtype;
        this.aIndex = aIndex;
        this.what = what;
        this.tupleDesc = tupleDesc;
        gbfield2agval = new HashMap<>();
    }

    /**
     * Merge a new tuple into the aggregate, grouping as indicated in the constructor
     * @param tup the Tuple containing an aggregate field and a group-by field
     */
    public void mergeTupleIntoGroup(Tuple tup) {
        // some code goes here
        Field gbField = null;
        Type gbfieldtype = null;
        if(Aggregator.NO_GROUPING!=gbIndex)
        {
            gbField = tup.getField(gbIndex);
            gbfieldtype = gbField.getType();
        }
        if(gbfieldtype!=Type.INT_TYPE)
        {
            throw new IllegalArgumentException();
        }
        Field aggreField = tup.getField(aIndex);
        if(!gbfield2agval.containsKey(gbField))
        {
            gbfield2agval.put(gbField,1);
        }
        else
        {
            int old_val = gbfield2agval.get(gbField);
            gbfield2agval.put(gbField,old_val+1);
        }

    }

    /**
     * Create a OpIterator over group aggregate results.
     *
     * @return a OpIterator whose tuples are the pair (groupVal,
     *   aggregateVal) if using group, or a single (aggregateVal) if no
     *   grouping. The aggregateVal is determined by the type of
     *   aggregate specified in the constructor.
     */
    public OpIterator iterator() {
        // some code goes here
        List<Tuple> tuples = new ArrayList<>();
        for (Map.Entry entry:gbfield2agval.entrySet())
        {
            Tuple tuple = new Tuple(tupleDesc);
            if(Aggregator.NO_GROUPING==gbIndex)
            {
                tuple.setField(0,new IntField((Integer) entry.getValue()));
            }
            else
            {
                tuple.setField(0, (Field) entry.getKey());
                tuple.setField(1, new IntField((Integer) entry.getValue()));
            }
            tuples.add(tuple);
        }
        return new TupleIterator(tupleDesc,tuples);
    }

}
